package com.cti.commons.delayqueue.handler;

import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.cti.commons.delayqueue.core.DelayBucket;
import com.cti.commons.delayqueue.core.DelayQueueJobPool;
import com.cti.commons.delayqueue.core.ReadyQueue;
import com.cti.commons.delayqueue.core.mode.ScoredSortedItem;

/**
 * @author et.wang
 * @description: 扫描延迟任务桶中的任务，放到准备队列中
 * @date 2022/10/27 19:34
 */
@Component
public class DelayBucketHandler {

    private static final Logger logger = LoggerFactory.getLogger(DelayBucketHandler.class);

    @Autowired
    private DelayBucket delayBucket;

    @Autowired
    private DelayQueueJobPool delayQueueJobPool;

    @Autowired
    private ReadyQueue readyQueue;

    //@SchedulerLock(name = "${delay.queue.bucketJobName:delayBucketHandler}",lockAtMostFor = "PT6s")
    public void execute(String delayBucketKey) {
        try {
            ScoredSortedItem item = delayBucket.getFromBucket(delayBucketKey);
            //没有任务
            if (item == null) {
                return;
            }
            //延迟时间没到
            if (item.getDelayTime() > System.currentTimeMillis()) {
                return;
            }
            logger.info("扫描延时桶key [{}],任务id [{}],topic[{}],时间[{}]",
                delayBucketKey, item.getDelayQueueJobId(), item.getTopic(), item.getDelayTime());
            readyQueue.pushToReadyQueue(item.getTopic(), item.getDelayQueueJobId());
            delayBucket.deleteFormBucket(delayBucketKey, item);
            logger.info("=================延时队列任务 [{},{}] 时间到，消息[{}]推送至待处理队列（新数据）=======================",
                delayBucketKey, item.getTopic(), item.getDelayQueueJobId());
        } catch (Exception e) {
            logger.error("扫描delayBucket出错：", e);
        }

    }

    private void sleep() {
        try {
            TimeUnit.SECONDS.sleep(1L);
        } catch (InterruptedException e) {
            logger.error("消费间隔出错", e);
        }
    }

}
